package com.xb.system.service.impl;

import com.google.common.hash.BloomFilter;
import com.xb.model.domain.Users;
import com.xb.system.constants.BatchConstants;
import com.xb.system.constants.ThreadPoolManager;
import com.xb.system.redis.UserPhoneBloomFilterConfig;
import com.xb.system.service.DynamicDataSplitterService;
import com.xb.system.service.FileHandler;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.io.BufferedReader;
import java.io.FileReader;
import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.BlockingQueue;

@Slf4j
@Component
public class TxtFileHandler implements FileHandler {

    @Value("${file.batch.size}")
    private int batchSize;

    @Resource
    private DynamicDataSplitterService dynamicDataSplitterService;

    @Resource
    private UserPhoneBloomFilterConfig userPhoneBloomFilterConfig;
    

    @Value("${file.thread.pool.size}")
    private int threadPoolSize;

    @Override
    public void handleFile(String file) throws IOException {
        initPool();
        BloomFilter<String> bloomFilter = userPhoneBloomFilterConfig.createBloomFilter();
        BlockingQueue<Users> queue = BatchConstants.QUEUE;
        BatchConstants.SIGNAL_COMPLETION.set(false);

        try (BufferedReader reader = new BufferedReader(new FileReader(file))) {
            String line;
            while ((line = reader.readLine()) != null) {
                List<Users> users = dynamicDataSplitterService.processLine(line.trim(), bloomFilter);
                if (users == null || users.isEmpty()) {
                    continue;
                }
                for (Users user : users) {
                    if (user != null) {
                        queue.put(user);
                    }
                }
            }
        } catch (IOException | InterruptedException e) {
            log.error("读取文件时发生错误：{}", e.getMessage());
            Thread.currentThread().interrupt();
        } finally {
            BatchConstants.SIGNAL_COMPLETION.set(true);
            log.info("入库完成");
            ThreadPoolManager.shutdown();
            log.info("线程池关闭");
        }
    }

    public void initPool() {
        ThreadPoolManager.getExecutor(threadPoolSize);
        for (int i = 0; i < threadPoolSize; i++) {
            ThreadPoolManager.submitTask(() -> BatchProcessor.processQueue(dynamicDataSplitterService, batchSize));
        }
    }
}

